package main

import (
	"fmt"
	"github.com/gogap/errors"
	"strconv"
	"strings"
	"time"

	"github.com/aliyunmq/mq-http-go-sdk"
)

var loopCount = 0

func ProcessError(err error) {
	// 如果Commit或Rollback时超过了TransCheckImmunityTime（针对发送事务消息的句柄）或者超过10s（针对consumeHalfMessage的句柄），则Commit或Rollback失败。
	if err == nil {
		return
	}
	fmt.Println(err)
	for _, errAckItem := range err.(errors.ErrCode).Context()["Detail"].([]mq_http_sdk.ErrAckItem) {
		fmt.Printf("\tErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s\n",
			errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
	}
}

func ConsumeHalfMsg(mqTransProducer *mq_http_sdk.MQTransProducer) {
	for {
		if loopCount >= 10 {
			return
		}
		loopCount++
		endChan := make(chan int)
		respChan := make(chan mq_http_sdk.ConsumeMessageResponse)
		errChan := make(chan error)
		go func() {
			select {
			case resp := <-respChan:
				{
					// 处理业务逻辑。
					var handles []string
					fmt.Printf("Consume %d messages---->\n", len(resp.Messages))
					for _, v := range resp.Messages {
						handles = append(handles, v.ReceiptHandle)
						fmt.Printf("\tMessageID: %s, PublishTime: %d, MessageTag: %s\n"+
							"\tConsumedTimes: %d, FirstConsumeTime: %d, NextConsumeTime: %d\n\tBody: %s\n"+
							"\tProperties:%s, Key:%s, Timer:%d, Trans:%d\n",
							v.MessageId, v.PublishTime, v.MessageTag, v.ConsumedTimes,
							v.FirstConsumeTime, v.NextConsumeTime, v.MessageBody,
							v.Properties, v.MessageKey, v.StartDeliverTime, v.TransCheckImmunityTime)

						a, _ := strconv.Atoi(v.Properties["a"])
						var comRollErr error
						if a == 1 {
							// 确认提交事务消息。
							comRollErr = (*mqTransProducer).Commit(v.ReceiptHandle)
							fmt.Println("Commit---------->")
						} else if a == 2 && v.ConsumedTimes > 1 {
							// 确认提交事务消息。
							comRollErr = (*mqTransProducer).Commit(v.ReceiptHandle)
							fmt.Println("Commit---------->")
						} else if a == 3 {
							// 确认回滚事务消息。
							comRollErr = (*mqTransProducer).Rollback(v.ReceiptHandle)
							fmt.Println("Rollback---------->")
						} else {
							// 什么都不做，下次再检查。
							fmt.Println("Unknown---------->")
						}
						ProcessError(comRollErr)
					}
					endChan <- 1
				}
			case err := <-errChan:
				{
					// Topic中没有消息可消费。
					if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
						fmt.Println("\nNo new message, continue!")
					} else {
						fmt.Println(err)
						time.Sleep(time.Duration(3) * time.Second)
					}
					endChan <- 1
				}
			case <-time.After(35 * time.Second):
				{
					fmt.Println("Timeout of consumer message ??")
					return
				}
			}
		}()

		// 长轮询检查半事务消息。
		// 长轮询表示如果Topic没有消息则请求会在服务端挂起3s，3s内如果有消息可以消费则立即返回响应。
		(*mqTransProducer).ConsumeHalfMessage(respChan, errChan,
			3, // 一次最多消费3条(最多可设置为16条)。
			3, // 长轮询时间3秒（最多可设置为30秒）。
		)
		<-endChan
	}
}

func main() {
	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
	endpoint := "http://1513085794924516.mqrest.cn-shenzhen.aliyuncs.com"
	// AccessKey ID阿里云身份验证，在阿里云RAM控制台创建。
	accessKey := "LTAI5tDBNA73QZaHNBSZiJQ8"
	// AccessKey Secret阿里云身份验证，在阿里云RAM控制台创建。
	secretKey := "xb7rgVSu9cJbELc9ZbzsuUj7eu7vpV"
	// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
	//不同消息类型的Topic不能混用，例如普通消息的Topic只能用于收发普通消息，不能用于收发其他类型的消息。
	topic := "dev_test_trans"
	// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
	instanceId := "MQ_INST_1513085794924516_BX63NFHI"
	// 您在控制台创建的Group ID。
	groupId := "GID_test"

	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")

	mqTransProducer := client.GetTransProducer(instanceId, topic, groupId)

	// 客户端需要有一个线程或者进程来消费没有确认的事务消息。
	// 启动一个Goroutines来检查没有确认的事务消息。
	go ConsumeHalfMsg(&mqTransProducer)

	// 发送4条事务消息，1条发送完就提交，其余3条通过检查半事务消息处理。
	//for i := 0; i < 1; i++ {
	//	iV64 := int64(i)
	//	iV32 := int32(i)
	//	iS := fmt.Sprintf("%d", i)
	//	md := mqmodel.NewStoreModel{
	//		Id:             iV64,
	//		//Name:           iS,
	//		//ProvinceCode:   iS,
	//		//CityCode:       iS,
	//		//DistrictCode:   iS,
	//		//DetailAddress:  iS,
	//		//Location:       iS,
	//		//Phone:          iS,
	//		//StoreType:      iV32,
	//		//Tags:           []string{iS},
	//		//PublicityImgs:  []string{iS},
	//		//RemarkNoti:     iS,
	//		//AreaSize:       iV32,
	//		//MaxPeople:      iV32,
	//		//ParkingLotType: iV64,
	//		//LockerType:     iV64,
	//		//ShowerRoomType: iV64,
	//		//Others:         []string{iS},
	//		//Owner:          0,
	//		//CreateTime:     time.Now(),
	//		//UpdateTime:     time.Now(),
	//		//Active:         false,
	//		//IsDel:          false,
	//		//MapShowName:    iS,
	//		//WifiType:       i,
	//	}
	//	str, err := jsonx.Marshal(&md)
	//	if err != nil {
	//		fmt.Println("Marshal error")
	//		return
	//	}
	//	msg := mq_http_sdk.PublishMessageRequest{
	//		MessageBody: string(str),
	//		Properties:  map[string]string{"a": strconv.Itoa(i)},
	//		MessageTag:  "store_loadEs",
	//	}
	//	// 设置事务第一次回查的时间，为相对时间。单位：秒，范围：10~300。
	//	// 第一次事务回查后如果消息没有Commit或者Rollback，则之后每隔10s左右会回查一次，共回查24小时。
	//	msg.TransCheckImmunityTime = 10
	//
	//	resp, pubErr := mqTransProducer.PublishMessage(msg)
	//	if pubErr != nil {
	//		fmt.Println(pubErr)
	//		return
	//	}
	//	fmt.Printf("Publish ---->\n\tMessageId:%s, BodyMD5:%s, Handle:%s\n",
	//		resp.MessageId, resp.MessageBodyMD5, resp.ReceiptHandle)
	//	//if i == 0 {
	//	// 发送完事务消息后能获取到半消息句柄，可以直接Commit或Rollback事务消息。
	//	ackErr := mqTransProducer.Commit(resp.ReceiptHandle)
	//	fmt.Println("Commit---------->")
	//	ProcessError(ackErr)
	//	//}
	//}

	for loopCount < 10 {
		time.Sleep(time.Duration(1) * time.Second)
	}
}
